package spark.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/2/10.
 * 该类写了大部分的DSL语法
 * 那么就会有个疑问，当真正开发的时候，到底是用DSL还是sparkCore还是sqlContext.sql()
 * sparkCore更多实现的是代码逻辑，
 *      如果你的逻辑比较强而且能够知道怎么写才更大化的优化内存和性能难么就使用sparkConte
 * 如果sql能力较好，可以直接使用sqlContext.sql()，这个是让代码自动帮你优化
 * 如果sql能力不佳，但是编码能力好，可以考虑DSL语法
 *      相对来说，你要考虑到更多的东西，而且这个语法不太好维护
 */

case class PerSon(empName:String,empAge:Int,empSex:String,salary:Int,deptNo:Int)
case class Dept(deptNo:Int,deptName:String)
object SparkSqlDSLDemo {
  def main(args: Array[String]) {
    //1、创建上下文
    val conf = new SparkConf()
          .setAppName("SparkSqlDSLDemo")
          .setMaster("local[*]")
          .set("spark.sql.shuffle.partitions","5")
    val sc = SparkContext.getOrCreate(conf)
    //因为要使用row_number窗口函数，所以使用hivecontext
    val sqlContext = new HiveContext(sc)
    //定义一个udf   sex：String  ==》 sex：Int，sexToNum
    sqlContext.udf.register("sexToNum",(sex:String) => {
      sex.toUpperCase match {
        case "M" => 0
        case "F" => 1
        case _ => -1
      }
    })
    sqlContext.udf.register("self_avg",Avg_UDAF)

    //2、加载数据
    /**
     *  RDD2DataFrame
     *  1、使用case class类型
     *  2、手动给定StructTpye
     */
    import sqlContext.implicits._
    val rdd1: RDD[PerSon] =
      sc.parallelize(Array(
        PerSon("张三",21,"M",1235,1),
        PerSon("李四",23,"M",1235,1),
        PerSon("王五",26,"F",1235,1),
        PerSon("赵六",19,"M",1235,1),
        PerSon("小花",21,"F",1225,1),
        PerSon("小华",30,"M",1515,2),
        PerSon("小林",45,"M",1565,2),
        PerSon("leo",17,"M",1865,2),
        PerSon("lili",34,"F",1915,2),
        PerSon("jack",41,"M",2543,3),
        PerSon("jack1",41,"XXXX",2543,2)
      ))
    val rdd2: RDD[Dept] =
      sc.parallelize(Array(
        Dept(1,"部门1"),
        Dept(2,"部门2"),
        Dept(4,"部门4")
      ))

    //将RDD转换成DataFrame的类型
    val personDataFrame = rdd1.toDF()
    val deptDataFrame = rdd2.toDF()

    //因为接下来会对这两个数据进行多次操作，所以cache
    personDataFrame.persist(StorageLevel.MEMORY_ONLY)
    deptDataFrame.persist(StorageLevel.MEMORY_ONLY)


    //==========================DSL语法====================================
    println("--------------select--------------------")

    personDataFrame.select("empName","empAge","empSex").show
    //使用隐式转换将String转换成Column
    personDataFrame.select($"empName".as("empName1"),$"empAge",$"empSex").show
    //当使用col来作为隐式转换，将StringtoColumn的时候，必须导入functions
    import org.apache.spark.sql.functions._
    personDataFrame.select(col("empName").as("empName2"),col("empAge"),col("empSex")).show
    //selectExpr在select的时候可以使用一些函数
    personDataFrame.selectExpr("empName","empAge","sexToNum(empSex) as sex_num").show


    println("--------------where/filter--------------------")
    //当使用where的时候，考虑后期维护优化的时候，代码尽量能够简单易懂
    personDataFrame.where("empAge > 22").where("empSex = 'M'").where("deptNo = 1").show
    personDataFrame.where("empAge > 22 AND empSex = 'M' AND deptNo = 1").show
    personDataFrame.where($"empAge" > 22 && ($"empSex" !== "F") &&  $"deptNo" === 1).show

    println("---------------------sort----------------------------")
    //全局排序
    //sort方法
    personDataFrame.sort("salary").select("empName","salary").show
    personDataFrame.sort($"salary".desc).select("empName","salary").show
    //先按照薪水进行降序排序，再按照年龄进行升序排序
    personDataFrame.sort($"salary".desc,$"empAge".asc)
      .select("empName","salary","empAge").show
    //orderby底层调用的是sort，所以不像hive当中有所区别
    println("============orderBy=============")
    personDataFrame.repartition(5).orderBy($"salary".desc)
      .select("empName","salary","empAge").show
    //局部排序
    println("========sortWithinPartitions==========")
    personDataFrame.repartition(5).sortWithinPartitions($"salary".desc,$"empAge".asc)
      .select("empName","salary","empAge").show

    //group by 聚合类的操作
    println("----------------------group by---------------------------")
    //按照性别分组，求平均薪水
    personDataFrame.groupBy("empSex")
      .agg(
        "salary" -> "avg"
      ).show
    //但是如上的用法会出现问题(前一个字段聚合出来的值，被后一个字段覆盖)
    // select avg(salary),sum(salary) from XXX group by empSex;
    personDataFrame.groupBy("empSex")
      .agg(
        "salary" -> "avg",
        "salary" -> "sum"
      ).show

    /**
     *  解决方法：
     */
    personDataFrame.groupBy("empSex")
      .agg(
        avg("salary").as("avg_salary"),
        sum("salary").as("sum_salary"),
        count(lit(1)).as("cnt")
      ).show

    /**
     * 如果要使用自定义的udaf只能传入map对象
     */
    personDataFrame.groupBy("empSex")
      .agg(
        "salary" -> "self_avg"
      ).show

    //limit
    println("---------------limit------------------")
    personDataFrame.limit(2).show


    //join
    println("--------------------------join------------------------------")

    /**
     * 分为左关联，右关联，内联，外联
     * a join b on a.XX = b.XX
     */
    //直接做join是内联（inner join）
    //如果直接使用join，其他参数什么都不给，那么就是做笛卡尔积，数据是有问题的
    //personDataFrame.join(deptDataFrame).show
    //无法判断deptNo到底属于哪张表，所以也会报错
    //personDataFrame.join(deptDataFrame,$"deptNo"===$"deptNo").show
    /**
     * 解决：
     *    方法一，换名称
     *     使用这个方法的时候，修改字段较少的那个DataFrame
     *     不然会写很多代码
     */
    personDataFrame.join(
    deptDataFrame.selectExpr("deptNo as col1","deptName"),
    $"deptNo"===$"col1"
    ).show
    personDataFrame.join(
      deptDataFrame.toDF("部门编号","deptName"),
      $"deptNo"===$"部门编号"
    ).show

    /**
     * 方法二，使用其他的join方法重载
     *
     */
    personDataFrame.join(deptDataFrame,"deptNo").show


    /**
     * 当出现两张表的两个列明一样的时候，
     * 做join不会出问题，
     * 但是接下来做操作就会报错
     * 解决的方法：
     *    事先改字段
     *    事后改字段
     */
    personDataFrame.join(
      deptDataFrame.toDF("deptNo","empName"),
      "deptNo"
    ).show
    //事后换名称的解决方法
    personDataFrame.join(
      deptDataFrame.toDF("deptNo","empName"),
      "deptNo"
    //|deptNo|empName|empAge|empSex|salary|empName|
    ).toDF("deptNo","empName","empAge","empSex","salary","deptName")
      .show

    personDataFrame.join(deptDataFrame,Seq("deptNo"),"left").show
    personDataFrame.join(deptDataFrame,Seq("deptNo"),"right").show()
    personDataFrame.join(deptDataFrame,Seq("deptNo"),"full").show


    println("-------------------case when的用法---------------------------")

    /**
     *  // Scala:
       people.select(when(people("gender") === "male", 0)
        .when(people("gender") === "female", 1)
        .otherwise(2))

   // Java:
      people.select(when(col("gender").equalTo("male"), 0)
        .when(col("gender").equalTo("female"), 1)
        .otherwise(2))
     */
    personDataFrame.select($"empName",$"empSex",when(personDataFrame("empSex") === "M","我是男生")
      .when(personDataFrame("empSex") === "F","我是女生")
      .otherwise("性别无法识别")
    ).show

    personDataFrame.select($"empName",$"empSex",
      when(personDataFrame("empSex") === "M" || personDataFrame("empSex") === "F","性别识别正常")
      .otherwise("性别识别异常")
      .as("识别")
    ).show


    println("--------------------窗口函数--------------------------")

    /**
     * 使用hive的窗口函数，就必须使用hivecontext
     *
     * 按照deptNo分组，组内按照salary进行排序，求前三salary
     *
     * 1、select *,ROW_NUMBER() OVER (PARTITION BY deptNo ODER BY salary desc) as rnk
     * 2、from Person      ===== tmp
     * 3、select * from tmp where rnk <= 3
     */
    //import org.apache.spark.sql.expressions.Window
    val w = Window.partitionBy("deptNo").orderBy($"salary".desc,$"empAge".asc)
    personDataFrame.select($"empName",
      $"empAge",
      $"deptNo",
      $"salary",
      row_number().over(w).as("rnk")
    ).where("rnk <= 3")
      .show
  }
}
